/*
 *  Filename:lsv_volume_gc.c
 *  Description:
 *
 *  Created on: 2017年3月13日
 *  Author: Asdf(825674301)
 */
#include "config.h"

#include <sys/types.h>
#include <unistd.h>

#define DBG_SUBSYS S_LIBSTORAGE

#include "dbg.h"
#include "lsv_help.h"
#include "lsv_volume.h"
#include "lsv_volume_gc.h"
#include "lsv_log.h"

#define print_vol_gc_info(vachk_list, type) \
        DINFO("<%u,%u>,<%u,%u>,%u,%u\n", (vachk_list)->malloc_record,\
                        ((vachk_list)->record + (vachk_list)->malloc_record)->count,\
                        (vachk_list)->free_record,\
                        ((vachk_list)->record + (vachk_list)->free_record)->count,\
                        (vachk_list)->next_chkid, (type));

#define volgc_stat(lsv_info) lsv_volume_gc_stat(lsv_info, __FUNCTION__, __LINE__)

static int lsv_volume_volgc_malloc_(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t *chunk_id);

static int lsv_volume_volgc_free_(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t chunk_id);


/**
 * @todo 512-align
 *
 * @param lsv_info
 * @return
 */
#if 0
static int __volume_gc_set_tail(lsv_volume_proto_t *lsv_info) {
        int ret = 0;
        lsv_volgc_info_t * volgc_info = NULL;
        int8_t buff[LSV_VOLGC_TAIL_AREA];
        lsv_u32_t *u32p, len;
        lsv_volume_io_t vio;

        volgc_info = lsv_info->volgc_info;

        len = 0;
        lsv_serialize(buff, len, lsv_u32_t, u32p, volgc_info->tail);

        YASSERT(len <= LSV_VOLGC_TAIL_AREA);

        lsv_volume_io_init(&vio, LSV_PRIM_CHUNK_ID,
                           lsv_info->u.volume_page_id * LSV_PAGE_SIZE,
                           len,
                           LSV_VOLGC_STORAGE_TYPE);
        ret = lsv_volume_chunk_update(lsv_info, &vio, buff);
        if (unlikely(ret)) {
                DFATAL("ret %d\n", ret);
                GOTO(err_ret, ret);
        }

        DINFO("set tail to %u successfully\n", volgc_info->tail);
        return 0;
err_ret:
        return ret;
}
#endif

static int __volume_gc_set_meta(lsv_volume_proto_t *lsv_info) {
        int ret;
        lsv_volgc_info_t *volgc_info = lsv_info->volgc_info;
        void *buff;//[LSV_PAGE_SIZE];

        ret = ymalloc(&buff, LSV_PAGE_SIZE);
        if(unlikely(ret))
                GOTO(err_ret, ret);

        memset(buff, 0, LSV_PAGE_SIZE);

        lsv_volume_meta_t *meta = (lsv_volume_meta_t *)buff;
        meta->tail = volgc_info->tail;

        for (int i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                meta->next_chkids[i] = volgc_info->avachk_list[i].next_chkid;
        }

        DINFO("set tail %u next_chkid %u %u\n",
              meta->tail,
              meta->next_chkids[0],
              meta->next_chkids[1]);

        lsv_volume_io_t vio;
        lsv_volume_io_init(&vio, LSV_PRIM_CHUNK_ID,
                           lsv_info->u.volume_page_id * LSV_PAGE_SIZE,
                           LSV_PAGE_SIZE,
                           LSV_VOLGC_STORAGE_TYPE);
        ret = lsv_volume_chunk_update(lsv_info, &vio, buff);
        if (unlikely(ret)) {
                // TODO IO error
                DFATAL("chunk_id %u off %u size %u ret %d\n", vio.chunk_id, vio.offset, vio.size, ret);
                GOTO(err_ret, ret);
        }

        yfree(&buff);
        return 0;
err_ret:
        return ret;
}

static int __init_volume_gc(lsv_volume_proto_t *lsv_info, int load_meta, int load_bitmap) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = NULL;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_volgc_avachk_record_t *avachk_record = NULL;
        void *buff;//[LSV_PAGE_SIZE];

        ret = ymalloc(&buff, LSV_PAGE_SIZE);
        if(unlikely(ret))
                return ret;

        YASSERT(LSV_VOLGC_TAIL_INIT == 65);
        YASSERT(LSV_STORAGE_TYPE_COUNT == 2);
        YASSERT(LSV_CHUNK_SIZE == sizeof(lsv_volgc_avachk_record_t));
        YASSERT(LSV_PAGE_SIZE > LSV_VOLGC_TAIL_AREA);

        // @malloc ERR1
        ret = ymalloc(&lsv_info->volgc_info, sizeof(lsv_volgc_info_t));
        if (unlikely(ret)) {
                ret = ENOMEM;
                DERROR("malloc failed, ret %d\n", ret);
                goto ERR1;
        }

        volgc_info = lsv_info->volgc_info;

        lsv_lock_init(&volgc_info->tail_lock);

        // create avachk_list each type
        for (int i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                avachk_list = &volgc_info->avachk_list[i];

                lsv_lock_init(&avachk_list->lock);
                avachk_list->malloc_record = 0;
                avachk_list->free_record = 0;
                avachk_list->next_chkid = LSV_CHUNK_NULL;

                avachk_list->malloc_count = 0;
                avachk_list->free_count = 0;

                for (int j = 0; j < LSV_VOLGC_AVACHK_LIST_NUM; j++) {
                        avachk_record = &avachk_list->record[j];
                        avachk_record->count = 0;
                        avachk_record->next_chkid = LSV_CHUNK_NULL;
                }
        }

        if (load_meta) {
                ret = lsv_volume_chunk_read_data(lsv_info, LSV_THIS_VOL_INO, LSV_PRIM_CHUNK_ID,
                                                 lsv_info->u.volume_page_id * LSV_PAGE_SIZE,
                                                 LSV_PAGE_SIZE, buff);
                if (unlikely(ret)) {
                        DERROR("upload volgc stroage err, ret %d\n", ret);
                        goto ERR1;
                }

                lsv_volume_meta_t *meta = (lsv_volume_meta_t *)buff;
                volgc_info->tail = meta->tail;

                // @hypo 应该为所有free list的
                // flush时，需要把free list的内存部分全部持久化
                // 如果不能满足以上假设，则需走recovery过程
                for (int i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                        avachk_list = &volgc_info->avachk_list[i];
                        avachk_list->next_chkid = meta->next_chkids[i];
                }
        } else {
                volgc_info->tail = LSV_VOLGC_TAIL_INIT;
                ret = __volume_gc_set_meta(lsv_info);
                if (unlikely(ret)) {
                        GOTO(ERR1, ret);
                }
        }

        DINFO("load tail %u next_chkid %u %u\n", volgc_info->tail,
              volgc_info->avachk_list[0].next_chkid,
              volgc_info->avachk_list[1].next_chkid);

        if (load_bitmap) {
                // 包含所有已分配出去的chunk_id
                for (int i = 0; i < LSV_VOLGC_STORAGE_CHUNK_NUM; i++) {
                        // chunk [1, 64]
                        ret = lsv_volume_chunk_read(lsv_info, LSV_VOLGC_STORAGE_SET_BASE + i,
                                                    (lsv_s8_t *)&volgc_info->bitmaps[i * LSV_CHUNK_SIZE]);
                        if (unlikely(ret)) {
                                DERROR("init bitmaps stroage err, ret %d\n", ret);
                                GOTO(ERR1, ret);
                        }
                }
        } else {
                memset(volgc_info->bitmaps, 0, LSV_VOLGC_STORAGE_BITMAP_NUM);
                for (int i = 0; i < LSV_VOLGC_STORAGE_CHUNK_NUM; i++) {
                        lsv_volume_io_t vio;
                        lsv_volume_io_init(&vio, LSV_VOLGC_STORAGE_SET_BASE + i, 0, LSV_CHUNK_SIZE, LSV_VOLGC_STORAGE_TYPE);
                        ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *)&volgc_info->bitmaps[i * LSV_CHUNK_SIZE]);
                        if (unlikely(ret)) {
                                DFATAL("init bitmaps stroage err, ret %d\n", ret);
                                GOTO(ERR1, ret);
                        }
                }
        }

        memset(volgc_info->dirty_page_bitmap, 0, LSV_VOLGC_STORAGE_BITMAP_DIRTY_BITMAP_NUM);

        yfree(&buff);
        return 0;
ERR1:
        yfree(&buff);
        lsv_volume_volgc_delete(lsv_info);
        return ret;
}

int lsv_volume_volgc_create(lsv_volume_proto_t *lsv_info) {
        int ret = 0;

        // @malloc ERR1
        ret = __init_volume_gc(lsv_info, 0, 0);
        if (unlikely(ret)) {
                GOTO(ERR0, ret);
        }

        return 0;
ERR0:
        return ret;
}

int lsv_volume_volgc_delete(lsv_volume_proto_t *lsv_info) {
        lsv_volgc_info_t *volgc_info = lsv_info->volgc_info;
        lsv_volgc_avachk_list_t *vachk_list = NULL;
        lsv_u32_t i;

        if (volgc_info) {
                lsv_lock_destroy(&volgc_info->tail_lock);
                for (i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                        vachk_list = volgc_info->avachk_list + i;
                        lsv_lock_destroy(&vachk_list->lock);
                }
                free(volgc_info);
        }
        lsv_info->volgc_info = NULL;
        return 0;
}

int lsv_volume_volgc_flush(lsv_volume_proto_t *lsv_info) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = lsv_info->volgc_info;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_volgc_avachk_record_t *new_record = NULL;

        DINFO("flush\n");

        for (int i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                avachk_list = volgc_info->avachk_list + i;
                lsv_lock(&avachk_list->lock);
                // TODO 如写入失败，则进入死循环，导致任务超时 schedule_yield
                while (1) {
                        new_record = &avachk_list->record[avachk_list->free_record];
                        DINFO("type %u malloc %u free %u count %u\n", i,
                              avachk_list->malloc_record,
                              avachk_list->free_record,
                              new_record->count);

                        if (new_record->count > 0) {
                                // 插入为单链表的头部
                                new_record->next_chkid = avachk_list->next_chkid;

                                // 为新record选择一chunk_id, 即其自身所包含的free chunk的第一个
                                // volume本身不能调用malloc机制，这种做法是可行的
                                avachk_list->next_chkid = new_record->avail_chkid[0];

                                lsv_volume_io_t vio;
                                lsv_volume_io_init(&vio, avachk_list->next_chkid, 0, LSV_CHUNK_SIZE, LSV_VOLGC_STORAGE_TYPE);
                                ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *)new_record);
                                if (unlikely(ret)) {
                                        avachk_list->next_chkid = new_record->next_chkid;
                                        new_record->next_chkid = LSV_CHUNK_NULL;
                                        DFATAL("write volume err, chk_id %u ret %d\n", avachk_list->next_chkid, ret);
                                        GOTO(err_ret, ret);
                                }
                        }

                        if (avachk_list->malloc_record == avachk_list->free_record) {
                                break;
                        }

                        // 从free到malloc，倒着走，符合stack方式
                        avachk_list->free_record = (avachk_list->free_record - 1 + LSV_VOLGC_AVACHK_LIST_NUM) % LSV_VOLGC_AVACHK_LIST_NUM;
                }
                lsv_unlock(&avachk_list->lock);
        }

        YASSERT(sizeof(lsv_volume_meta_t) == sizeof(uint32_t) * (LSV_STORAGE_TYPE_COUNT + 1));

        ret = __volume_gc_set_meta(lsv_info);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        lsv_volume_volgc_delete(lsv_info);
        return 0;
err_ret:
        lsv_volume_volgc_delete(lsv_info);
        return ret;
}

int lsv_volume_volgc_load(lsv_volume_proto_t *lsv_info) {
        int ret = 0;

        DINFO("load\n");

        // @malloc ERR1
        ret = __init_volume_gc(lsv_info, 1, 1);
        if (unlikely(ret)) {
                GOTO(ERR0, ret);
        }

        return 0;
ERR0:
        return ret;
}

int lsv_volume_volgc_recovery(lsv_volume_proto_t *lsv_info) {
        int ret = 0;

        LSV_TEST_TIME_BEGIN(volume_recovery1);

        // @malloc ERR1
        ret = __init_volume_gc(lsv_info, 1, 1);
        if (unlikely(ret)) {
                GOTO(ERR0, ret);
        }

        // recovery
        LSV_TEST_TIME_END(volume_recovery1);

        lsv_volgc_info_t *volgc_info = lsv_info->volgc_info;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_u8_t type;

        for (int i = 0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                // TODO 造成垃圾
                avachk_list = &volgc_info->avachk_list[i];
                avachk_list->next_chkid = LSV_CHUNK_NULL;
        }

        DINFO("volgc recovery begin\n");
        LSV_TEST_TIME_BEGIN(volume_recovery2);

        int count = 0;
        int free_count = 0;
        // 检查所有的底层chunk，是否已分配，如没有，则回收
        for (int i = LSV_VOLGC_TAIL_INIT; i < volgc_info->tail; i++) {
                type = (volgc_info->bitmaps[i / (8 / LSV_VOLGC_STORAGE_BIT_LEN)]
                        >> (i % (8 / LSV_VOLGC_STORAGE_BIT_LEN) * LSV_VOLGC_STORAGE_BIT_LEN)) & ((lsv_u8_t) 0x03);

                if (count % 1000 == 0) {
                        DINFO("check bitmaps count %d chunk_id:%u,type:%u\n", count, i, type);
                }
                count++;

                // 没被使用, 高位为0
                // 两位中，高位表示是否占用，低位0：SSD，1：HDD
                if (!(type & ((lsv_u8_t) 0x02))) {
                        if (type == 0) {
                                type = LSV_STORAGE_TYPE_SSD;
                        } else {
                                type = LSV_STORAGE_TYPE_HDD;
                        }
                        DINFO("revocery bitmaps chunk_id:%u, type:%u\n", i, type);
                        free_count++;
                        lsv_volume_volgc_free_(lsv_info, type, i);
                }
        }

        LSV_TEST_TIME_END(volume_recovery2);
        DINFO("volgc recovery end, tail %u free %u\n", volgc_info->tail, free_count);

        // must set 0 after recovery
        memset(volgc_info->dirty_page_bitmap, 0, LSV_VOLGC_STORAGE_BITMAP_DIRTY_BITMAP_NUM);

        return 0;
ERR0:
        return ret;
}

/**
 * @todo linear allocator，可以与disk分配配合，提高顺序写入性能
 *
 * @note avachk就是free list，目前按队列和stack的混合模式组织，具有很大的随机性
 * @note free过程是随机的，导致free list的组织并不是有序的
 *
 * @param lsv_info
 * @param type
 * @param chunk_id
 * @return
 */
static int lsv_volume_volgc_malloc_(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t *chunk_id) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = NULL;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_volgc_avachk_record_t *avachk_record = NULL;
        lsv_u32_t malloc_chk_num, chkid, max_tail;
        lsv_u32_t i;

        assert(type < LSV_STORAGE_TYPE_COUNT);

        volgc_info = lsv_info->volgc_info;
        avachk_list = &volgc_info->avachk_list[type];
        print_vol_gc_info(avachk_list, type);

        // 每类可用chunk列表按环形队列组织
        avachk_record = &avachk_list->record[avachk_list->malloc_record];

        // not enough to extend
        if (avachk_list->malloc_record == avachk_list->free_record) {
                // is empty
                if (avachk_record->count == 0) {
                        if (LSV_CHUNK_NULL != avachk_list->next_chkid) {
                                // load from storage
                                ret = lsv_volume_chunk_read(lsv_info, avachk_list->next_chkid, (lsv_s8_t *)avachk_record);
                                if (unlikely(ret)) {
                                        DERROR("read volume err, chk_id %u ret %d\n", avachk_list->next_chkid, ret);
                                        goto ERR;
                                }
                                avachk_list->next_chkid = avachk_record->next_chkid;
                                avachk_record->next_chkid = LSV_CHUNK_NULL;
                                print_vol_gc_info(avachk_list, type);
                        } else {
                                // TODO extend volume tail
                                max_tail = (uint32_t) (lsv_info->max_size / LSV_CHUNK_SIZE);
                                lsv_lock(&volgc_info->tail_lock);
                                if (volgc_info->tail < max_tail) {
                                        malloc_chk_num = _min(LSV_VOLGC_RECORD_AVACHK_NUM - avachk_record->count,
                                                              LSV_VOLGC_MALLOC_CHUNK_NUM);
                                        malloc_chk_num = _min(malloc_chk_num, max_tail - volgc_info->tail);
                                        DINFO("change tail from %u to %u, malloc_chk_num %u max_tail %u\n",
                                              volgc_info->tail,
                                              volgc_info->tail + malloc_chk_num,
                                              malloc_chk_num,
                                              max_tail);

                                        // extend
                                        volgc_info->tail += malloc_chk_num;
                                        ret = __volume_gc_set_meta(lsv_info);
                                        if (unlikely(ret)) {
                                                GOTO(ERR, ret);
                                        }

                                        chkid = volgc_info->tail;
                                        lsv_unlock(&volgc_info->tail_lock);

                                        // used as a stack, push to stack top
                                        for (i = 0; i < malloc_chk_num; i++) {
                                                avachk_record->avail_chkid[avachk_record->count++] = --chkid;
                                        }
                                        print_vol_gc_info(avachk_list, type);
                                } else {
                                        //storage end
                                        lsv_unlock(&volgc_info->tail_lock);
                                        if (avachk_record->count == 0) {
                                                ret = ENOMEM;
                                                DFATAL("no enough space, ret %d tail %u vol_size %ju\n", ret,
                                                       volgc_info->tail,
                                                       lsv_info->max_size);
                                                GOTO(ERR, ret);
                                        }
                                        print_vol_gc_info(avachk_list, type);
                                }
                        }
                }

                // 重合时
                avachk_list->free_record = (avachk_list->free_record + 1) % LSV_VOLGC_AVACHK_LIST_NUM;
        } else {
                YASSERT(avachk_record->count > 0);
        }

        // malloc a chunk, pop stack top
        chkid = avachk_record->avail_chkid[--avachk_record->count];
        if (avachk_record->count == 0) {
                avachk_list->malloc_record = (avachk_list->malloc_record + 1) % LSV_VOLGC_AVACHK_LIST_NUM;
        }

        print_vol_gc_info(avachk_list, type);
        lsv_volume_bitmap_malloc(lsv_info, chkid, type);

        YASSERT(chkid != LSV_CHUNK_NULL && chkid < volgc_info->tail);

        DINFO("malloc chunk_id %u\n", chkid);

        *chunk_id = chkid;
        return 0;
ERR:
        return ret;
}

static int lsv_volume_volgc_free_(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t chunk_id) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = NULL;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_volgc_avachk_record_t *avachk_record = NULL;
        lsv_u32_t next_free_record;

        assert(type < LSV_STORAGE_TYPE_COUNT);

        if (LSV_CHUNK_NULL == chunk_id) {
                DERROR("can't free first chunk,errno:%d,chunk id:%u\n", ret, chunk_id);
                ret = EPERM;
                goto ERR;
        }

        volgc_info = lsv_info->volgc_info;
        avachk_list = volgc_info->avachk_list + type;
        print_vol_gc_info(avachk_list, type);

        avachk_record = avachk_list->record + avachk_list->free_record;

        // full, select free chunk (reuse current or use next circularly)
        if (avachk_record->count == LSV_VOLGC_RECORD_AVACHK_NUM) {
                next_free_record = (avachk_list->free_record + 1) % LSV_VOLGC_AVACHK_LIST_NUM;
                if (next_free_record == avachk_list->malloc_record) {
                        // down to storage
                        avachk_record->next_chkid = avachk_list->next_chkid;
                        // TODO volume层不能直接调用malloc, use the first free chunk
                        avachk_list->next_chkid = avachk_record->avail_chkid[0];

                        lsv_volume_io_t vio;
                        lsv_volume_io_init(&vio, avachk_list->next_chkid, 0, LSV_CHUNK_SIZE, LSV_VOLGC_STORAGE_TYPE);
                        ret = lsv_volume_chunk_update(lsv_info, &vio, (lsv_s8_t *)avachk_record);
                        if (unlikely(ret)) {
                                avachk_list->next_chkid = avachk_record->next_chkid;
                                avachk_record->next_chkid = LSV_CHUNK_NULL;
                                DFATAL("write volume err, ret %d chk_id %u\n", ret, avachk_list->next_chkid);
                                goto ERR;
                        }
                        avachk_record->count = 0; // set in free
                        avachk_record->next_chkid = LSV_CHUNK_NULL;
                        print_vol_gc_info(avachk_list, type);
                } else {
                        //next
                        avachk_list->free_record = next_free_record;
                        avachk_record = avachk_list->record + avachk_list->free_record;
                        print_vol_gc_info(avachk_list, type);
                }
        }

        avachk_record->avail_chkid[avachk_record->count] = chunk_id;
        avachk_record->count++;
        print_vol_gc_info(avachk_list, type);
        lsv_volume_bitmap_free(lsv_info, chunk_id, type);

        DINFO("free chunk_id %u\n", chunk_id);
ERR:
        return ret;
}

int lsv_volume_volgc_malloc(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t *chunk_id) {
        return lsv_volume_volgc_malloc_batch(lsv_info, type, 1, chunk_id);
}

int lsv_volume_volgc_free(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t chunk_id) {
        return lsv_volume_volgc_free_batch(lsv_info, type, 1, &chunk_id);
}

int lsv_volume_volgc_malloc_batch(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t count, lsv_u32_t *chunk_id) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = NULL;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_u32_t i;

        volgc_info = lsv_info->volgc_info;
        avachk_list = &volgc_info->avachk_list[type];

        // TODO core timeout
        // @malloc ERR1
        lsv_lock(&avachk_list->lock);

        for (i = 0; i < count; i++) {
                ret = lsv_volume_volgc_malloc_(lsv_info, type, &chunk_id[i]);
                if (unlikely(ret)) {
                        DERROR("malloc chunk fail %u/%u, ret %d\n", i, count, ret);
                        memset(volgc_info->dirty_page_bitmap, 0, LSV_VOLGC_STORAGE_BITMAP_DIRTY_BITMAP_NUM); //clean
                        GOTO(ERR1, ret);
                }

                DINFO("%u/%u type %u chunk_id %u\n", i, count, type, chunk_id[i]);
        }

        ret = lsv_volume_bitmap_batch_set(lsv_info, count, chunk_id);
        if (unlikely(ret)) {
                GOTO(ERR1, ret);
        }

        avachk_list->malloc_count += count;

        lsv_unlock(&avachk_list->lock);

        volgc_stat(lsv_info);
        return 0;
ERR1:
        lsv_unlock(&avachk_list->lock);
        return ret;
}

int lsv_volume_volgc_free_batch(lsv_volume_proto_t *lsv_info, lsv_u8_t type, lsv_u32_t count, lsv_u32_t *chunk_id) {
        int ret = 0;
        lsv_volgc_info_t *volgc_info = NULL;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        lsv_u32_t i;

        volgc_info = lsv_info->volgc_info;
        avachk_list = volgc_info->avachk_list + type;

        // @malloc ERR1
        lsv_lock(&avachk_list->lock);

        for (i = 0; i < count; i++) {
                ret = lsv_volume_volgc_free_(lsv_info, type, chunk_id[i]);
                if (unlikely(ret)) {
                        DERROR("free chunk %u/%u, ret %u\n", i, count, ret);
                        memset(volgc_info->dirty_page_bitmap, 0, LSV_VOLGC_STORAGE_BITMAP_DIRTY_BITMAP_NUM); //clean
                        goto ERR1;
                }

                DINFO("%u/%u type %u chunk_id %u\n", i, count, type, chunk_id[i]);
        }

        ret = lsv_volume_bitmap_batch_set(lsv_info, count, chunk_id);
        if (unlikely(ret)) {
                GOTO(ERR1, ret);
        }

        avachk_list->free_count += count;
        lsv_unlock(&avachk_list->lock);

        volgc_stat(lsv_info);
        return 0;
ERR1:
        lsv_unlock(&avachk_list->lock);
        return ret;
}

int __volgc_avachk_list_stat(lsv_volgc_avachk_list_t *avachk_list, uint32_t *chunk_num) {
        uint32_t i;
        lsv_volgc_avachk_record_t *record;

        *chunk_num = 0;

        i = avachk_list->malloc_record;
        while (1) {
                record = &avachk_list->record[i];
                *chunk_num += record->count;
                DINFO("idx %u count %u total %u next_chkid %u\n", i, record->count, *chunk_num, record->next_chkid);
                if (i == avachk_list->free_record) {
                        break;
                }

                i = (i + 1) % LSV_VOLGC_AVACHK_LIST_NUM;
        }

        return 0;
}

int lsv_volume_gc_stat(lsv_volume_proto_t *lsv_info, const char *func, int line) {
        lsv_volgc_info_t *volgc_info = lsv_info->volgc_info;
        lsv_volgc_avachk_list_t *avachk_list = NULL;
        uint32_t chunk_num;

        for (int i=0; i < LSV_STORAGE_TYPE_COUNT; i++) {
                avachk_list = &volgc_info->avachk_list[i];
                __volgc_avachk_list_stat(avachk_list, &chunk_num);
                DINFO("[%s, %d] tail %u type %u idx %u %u chunk_num %u count %ju %ju next_chkid %u\n", func, line,
                      volgc_info->tail,
                      i,
                      avachk_list->malloc_record,
                      avachk_list->free_record,
                      chunk_num,
                      avachk_list->malloc_count,
                      avachk_list->free_count,
                      avachk_list->next_chkid);
        }

        return 0;
}
